package com.linkstec.kafka;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaException;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.linkstec.util.ConfigUtil;

public class KafkaSender {

	private static Properties props;

	static {
		props = new Properties();
		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
				ConfigUtil.getProperty("bootstrap.servers", "storm01:9093,storm02:9093,storm03:9093"));
		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
				ConfigUtil.getProperty("key_serializer", "org.apache.kafka.common.serialization.StringSerializer"));
		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
				ConfigUtil.getProperty("value_serializer", "org.apache.kafka.common.serialization.StringSerializer"));
		props.put(ProducerConfig.ACKS_CONFIG,
				ConfigUtil.getProperty("acks", "all"));
//		props.put(ProducerConfig.TIMEOUT_CONFIG,
//				ConfigUtil.getProperty("timeout.ms", "3000"));
//		props.put(ProducerConfig.METADATA_FETCH_TIMEOUT_CONFIG,
//				ConfigUtil.getProperty("metadata.fetch.timeout.ms", "3000"));
		props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
				ConfigUtil.getProperty("compression.type", "none"));
		props.put(ProducerConfig.RETRIES_CONFIG,
				ConfigUtil.getProperty("retries", "3"));
		props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG,
				ConfigUtil.getProperty("reconnect.backoff.ms", "100"));
		props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 1024 * 1024 * 1024);
		props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
		props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);

	}

/*	private static ThreadLocal<KafkaProducer<String, String>> threadLocal = new ThreadLocal<KafkaProducer<String, String>>() {
		public KafkaProducer<String, String> initialValue() {
			return new KafkaProducer<String, String>(props);
		}
	};
*/
	static class innerStaticClass{
       private static KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);
    }
	public static KafkaProducer<String, String> getKafkaProducer() {
		KafkaProducer<String, String> producer = innerStaticClass.kafkaProducer;

		return producer;

	}

	public static void sendKafkaMessage(String topic, Object kafkamsg)
			throws Exception {
		KafkaProducer<String, String> localProducer = getKafkaProducer();
		String message = null;
		if (!(kafkamsg instanceof String)) {
			//message = JSONObject.fromObject(kafkamsg).toString();
			message = JSONArray.toJSONString(kafkamsg);
		} else {
			message = (String) kafkamsg;
		}

		ProducerRecord<String, String> msg = new ProducerRecord<String, String>(
				topic, Integer.toString(message.hashCode()), message);
		try {
			Future<RecordMetadata> metaDataFuture = localProducer.send(msg);
			metaDataFuture.get();
		} catch (KafkaException e) {
			throw e;
		} catch (InterruptedException e1) {
			throw e1;
		} catch (ExecutionException e2) {
			throw e2;
		}
	}

	public static void sendKafkaMessageAsync(String topic, Object kafkamsg)
			throws Exception {
		KafkaProducer<String, String> localProducer = getKafkaProducer();
		String message = null;
		if (!(kafkamsg instanceof String)) {
			message = JSON.toJSONString(kafkamsg, SerializerFeature.WriteClassName);
		} else {
			message = (String) kafkamsg;
		}

        ProducerRecord<String, String> msg = new ProducerRecord<String, String>(topic,
                Integer.toString(message.hashCode()), message);
		localProducer.send(msg, new Callback() {
			@Override
			public void onCompletion(RecordMetadata metaData, Exception exception) {
				if (exception != null) {
					if (metaData != null) {
					}
				}
			}
		});
	}
	public static void close(KafkaProducer<String, String> producer) {
		if (producer != null) {
			producer.close();
		}
	}

	public static void main(String[] args) {
		try {
			KafkaSender.sendKafkaMessage(ConfigUtil.getProperty("topic", "mot_test"),
					"Hello World 1!");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}